package de.akabahn.ebd.comm.dbd;

import static de.akabahn.ebd.comm.Util.assertInRange;
import static de.akabahn.ebd.comm.Util.assertNotNull;
import static de.akabahn.ebd.comm.Util.waitForLatch;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import de.akabahn.ebd.comm.ConnectionException;
import de.akabahn.ebd.comm.ConnectionObserver;
import de.akabahn.ebd.comm.ObservableConnection;
import de.akabahn.ebd.comm.Util;
import de.akabahn.ebd.comm.socket.SocketBridge;
import de.akabahn.ebd.comm.telnet.TelnetConnectionClient;

/**
 * Client implementation for a connection to the DBD server using the binary
 * protocol for the communication.<br>
 * For the text protocol, the <code>TelnetConnectionClient</code> class should
 * be used.<br>
 * Extends from <code>ObservableConnection</code> so any class that is
 * interested in the current connection status and in the change of this status,
 * may register itself as an observer so it gets notified on a connection state
 * change.<br>
 * Reading from the DBD server is done asynchronously in a separate thread.
 * 
 * @author Peter Eimann
 * @since 0.2
 * @version 0.3
 * @see TelnetConnectionClient
 * @see DBDParameter
 * @see DBDCommand
 * @see SocketBridge
 */
public class DBDConnectionClient extends ObservableConnection implements Runnable {

	private final static Logger LOG = Logger.getLogger(DBDConnectionClient.class);

	/** The DBD command to end the session */
	private final static DBDParameter END_OF_SESSION = new DBDParameter(DBDCommand.ENDOFSESSION, "TEST");

	private String server;
	private int port;

	private Socket socket;
	private InputStream inputStream;
	private OutputStream outputStream;
	private ConnectionObserver connectionObserver;
	private SocketBridge socketBridge;
	private DBDInputListener inputListener;

	private ExecutorService executor;

	private boolean stopped;

	/**
	 * Create and configure the client. To connect to the DBD server, you have
	 * to call <code>connect()</code> first.
	 * 
	 * @param server
	 *            The server to connect to
	 * @param port
	 *            The port to connect to
	 * @throws IllegalArgumentException
	 *             thrown if the input parameters are invalid
	 * @see #connect()
	 */
	public DBDConnectionClient(String server, int port) {
		assertNotNull(server, "server");
		assertInRange(port, "port", 0, 65535);

		this.server = server;
		this.port = port;

		socket = new Socket();
		stopped = true;
	}

	/**
	 * Register the <code>DBDInputListener</code> that is called when data has
	 * been received from the DBD server.
	 * 
	 * @param inputListener
	 *            The input listener instance
	 */
	public void registerInputListener(DBDInputListener inputListener) {
		assertNotNull(inputListener, "inputListener");
		this.inputListener = inputListener;
	}

	/**
	 * Unregister the current <code>DBDInputListener</code>. Data that will be
	 * received from the DBD server after this call will not be sent to a
	 * client.
	 */
	public void unregisterInputListener() {
		inputListener = null;
	}

	/**
	 * Connect to the DBD server using the socket bridge. If the connection
	 * could be established, the <code>ConnectionObserverThread</code> is
	 * started to observe changes in the connection status and the
	 * <code>InputListenerThread</code> is started to be able to read data from
	 * the DBD server.
	 * 
	 * @throws ConnectionException
	 * @see SocketBridge
	 * @see ConnectionObserver
	 */
	public synchronized void connect() throws ConnectionException {
		if (stopped) {
			CountDownLatch latch = new CountDownLatch(1);

			stopped = false;
			executor = Executors.newCachedThreadPool();

			if (socketBridge == null) {
				socketBridge = new SocketBridge(socket, server, port, latch);
				executor.execute(socketBridge);

				if (!waitForLatch(latch, 30000)) {
					throw new ConnectionException("Connect timed out");
				}

				if (socketBridge.getThreadException() != null) {
					throw new ConnectionException("Connect failed", socketBridge.getThreadException());
				}
				
				try {
					inputStream = socket.getInputStream();
					outputStream = socket.getOutputStream();
				} catch (IOException e) {
					throw new ConnectionException("Get input or output stream failed", e);
				}
			}

			if (connectionObserver == null) {
				connectionObserver = new ConnectionObserver(this);
				executor.execute(connectionObserver);
			}

			executor.execute(this);
		}
	}

	/**
	 * Close the current connection by stopping the
	 * <code>SocketBridge</code> thread, the
	 * <code>ConnectionObserver</code> thread and the
	 * <code>InputListener</code> thread.
	 * 
	 * @throws ConnectionException
	 */
	public synchronized void disconnect() throws ConnectionException {
		if (executor != null && !stopped) {
			if (socketBridge != null) {
				write(END_OF_SESSION);
				socketBridge.prepareToStop();
			}
			if (connectionObserver != null) {
				connectionObserver.prepareToStop();
			}

			stopped = true;
			executor.shutdownNow();
			try {
				if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
					throw new ConnectionException(
							"At lease one thread is still alive", socketBridge.getThreadException());
				}
			} catch (InterruptedException e) {
			} finally {
				socketBridge = null;
				connectionObserver = null;
				executor = null;
			}
		}
	}

	/**
	 * Checks if a connection to the server exists or not by calling
	 * <code>isConnected()</code> on the <code>Socket</code>.
	 * 
	 * @return <code>true</code> if a connection exists, <code>false</code>
	 *         otherwise
	 */
	@Override
	public boolean isConnected() {
		return socket.isConnected() && !socket.isClosed();
	}

	/**
	 * Reestablish the server connection by calling <code>disconnect()</code>
	 * and <code>connect()</code>.
	 */
	@Override
	public synchronized void reconnect() {
		try {
			disconnect();
		} catch (RuntimeException e) {
			// Ignore exceptions (maybe was already disconnected)
		}
		connect();
	}

	/**
	 * Send the passed <code>DBDParameter</code> to the DBD server, according to
	 * the specification of the binary protocol (21 bytes):<br>
	 * <ol>
	 * <li>4 bytes values, little endian</li>
	 * <li>1 byte command</li>
	 * <li>16 bytes name of the variable</li>
	 * </ol>
	 * 
	 * @param dbd
	 *            The data that will be sent to the DBD server
	 * @throws ConnectionException
	 * @throws IllegalArgumentException
	 *             thrown if the input parameters are invalid
	 */
	public synchronized void write(DBDParameter dbd) throws ConnectionException {
		assertNotNull(dbd, "dbd");
		assertNotNull(dbd.getCommand(), "dbd.command");
		assertNotNull(dbd.getName(), "dbd.name");
		byte[] writeBuffer = new byte[21];

		// value
		ByteBuffer valueBuffer = ByteBuffer.allocate(4);
		valueBuffer.order(ByteOrder.LITTLE_ENDIAN);
		valueBuffer.putInt(dbd.getValue());
		System.arraycopy(valueBuffer.array(), 0, writeBuffer, 0, 4);

		// command
		writeBuffer[4] = dbd.getCommand().getValue();

		// variable
		System.arraycopy(dbd.getName().getBytes(), 0, writeBuffer, 5, dbd.getName().length());

		try {
			LOG.debug("> " + dbd);
			outputStream.write(writeBuffer);
			outputStream.flush();
		} catch (Exception e) {
			throw new ConnectionException("Write failed", e);
		}
	}

	/**
	 * Return a string representation of this instance
	 */
	public String toString() {
		return "DBDConnection [IP=" + server + ", PO=" + port + ", CO=" + isConnected() + "]";
	}

	/**
	 * Loop that runs in a separate thread that reads data from the DBD server
	 * and that calls <code>onInputAvailable()</code> on the registered
	 * <code>DBDInputListener</code> if data has been received.<br>
	 * The loop continues until the stopped flag is set to <code>true</code>.
	 * 
	 * @see DBDInputListener#onInputAvailable(DBDParameter)
	 */
	public void run() {
		DBDParameter dbd = null;

		while (!stopped) {
			try {
				dbd = read();
				if (inputListener != null) {
					inputListener.onInputAvailable(dbd);
				}
			} catch (RuntimeException e) {
				if (!stopped) {
					LOG.error("onInputAvailable:", e);
					Util.sleep(500);
				}
			}
		}
	}

	/**
	 * Read the response from the DBD server according to the specification of
	 * the binary protocol (21 bytes):<br>
	 * <ol>
	 * <li>4 bytes values, little endian</li>
	 * <li>1 byte command</li>
	 * <li>16 bytes name of the variable</li>
	 * </ol>
	 * This method operates in a non-blocking mode, before <code>read()</code>
	 * is called on the input stream, it's checked if data is available and if
	 * not, this method returns <code>null</code>.
	 * 
	 * @return The response from the DBD server (will be <code>null</code> if no
	 *         data is available)
	 * @throws ConnectionException
	 */
	DBDParameter read() throws ConnectionException {
		byte[] readBuffer = new byte[21];
		int in = 0;
		int pos = 0;

		try {
			while ((pos < readBuffer.length) && ((in = inputStream.read()) != -1)) {
				readBuffer[pos++] = (byte) in;
			}
		} catch (Exception e) {
			throw new ConnectionException("Read failed", e);
		}

		ByteBuffer valueBuffer = ByteBuffer.wrap(readBuffer, 0, 4);
		valueBuffer.order(ByteOrder.LITTLE_ENDIAN);

		DBDCommand command = DBDCommand.fromValue(readBuffer[4]);
		String name = new String(readBuffer, 5, 16).replace((char) 0x00, ' ');
		int value = valueBuffer.getInt();
		DBDParameter dbd = new DBDParameter(command, name.trim(), value);
		LOG.debug("< " + dbd);

		return dbd;
	}
}
